import io.etcd.jetcd.KeyValue;
import io.etcd.jetcd.watch.WatchResponse;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.nio.charset.Charset;
import java.util.function.Consumer;

public class MyConsumer implements Consumer<WatchResponse> {
    private SourceFunction.SourceContext<String> ctx;

    public MyConsumer(SourceFunction.SourceContext<String> ctx) {
        this.ctx = ctx;
    }

    public MyConsumer() {
    }

    @Override
    public void accept(WatchResponse watchResponse) {
        KeyValue keyValue = watchResponse.getEvents().iterator().next().getKeyValue();
        String res = "etcd 's config 's config key is :{" + keyValue.getKey().toString(Charset.defaultCharset()) + "},value is:{" + keyValue.getValue().toString(Charset.defaultCharset()) + "}";
        if (ctx == null) {
            System.out.println(res);
        } else {
            ctx.collect(res);
        }
    }
}
